//package com.informix.stream.transactions;
//
//
//import com.informix.stream.api.*;
//import com.informix.stream.cdc.records.IfxCDCBeginTransactionRecord;
//import com.informix.stream.common.exception.IfxStreamException;
//import org.slf4j.Logger;
//import org.slf4j.LoggerFactory;
//
//import java.sql.SQLException;
//import java.util.*;
//import java.util.concurrent.ConcurrentHashMap;
//
//public class IfmxTransactionEngine
//        implements IfxTransactionEngine {
//    private final Logger logger = LoggerFactory.getLogger(IfmxTransactionEngine.class);
//    protected static final Set<IfmxStreamRecordType> ALLOWED_DATA_FILTERS = EnumSet.of(IfmxStreamRecordType.INSERT, IfmxStreamRecordType.DELETE, IfmxStreamRecordType.BEFORE_UPDATE, IfmxStreamRecordType.AFTER_UPDATE);
//
//
//    protected static final Set<IfmxStreamRecordType> ALLOWED_TRANSACTION_FILTERS = EnumSet.of(IfmxStreamRecordType.COMMIT, IfmxStreamRecordType.ROLLBACK);
//
//
//    protected EnumSet<IfmxStreamRecordType> dataFilterSet = EnumSet.of(IfmxStreamRecordType.INSERT, IfmxStreamRecordType.DELETE, IfmxStreamRecordType.BEFORE_UPDATE, IfmxStreamRecordType.AFTER_UPDATE);
//
//
//    protected EnumSet<IfmxStreamRecordType> transactionFilterSet = EnumSet.of(IfmxStreamRecordType.COMMIT);
//    protected final Map<Integer, TransactionHolder> transactionMap = new ConcurrentHashMap<>();
//
//    protected boolean sendEmptyTransactions = false;
//    protected final IfmxStreamEngine engine;
//
//    public IfmxTransactionEngine(IfmxStreamEngine engine) {
//        this.engine = engine;
//    }
//
//    public static class TransactionHolder {
//        public IfxCDCBeginTransactionRecord beginRecord;
//        public IfmxStreamRecord closingRecord;
//        public final List<IfmxStreamOperationRecord> records = new ArrayList<>();
//    }
//
//
//    public IfmxStreamTransactionRecord getTransaction() throws SQLException, IfxStreamException {
//        IfmxStreamRecord r;
//        while ((r = this.engine.getRecord()) != null) {
//
//            if (r.getType() == IfmxStreamRecordType.BEGIN) {
//                TransactionHolder h = new TransactionHolder();
//                h.beginRecord = (IfxCDCBeginTransactionRecord) r;
//                this.transactionMap.put(Integer.valueOf(r.getTransactionId()), h);
//                this.logger.debug("Watching trasaction id: {}", Integer.valueOf(r.getTransactionId()));
//                continue;
//            }
//            if (this.transactionMap.containsKey(Integer.valueOf(r.getTransactionId()))) {
//                TransactionHolder h = this.transactionMap.get(Integer.valueOf(r.getTransactionId()));
//                processRecord(r, h);
//                if (h.closingRecord != null) {
//                    this.transactionMap.remove(Integer.valueOf(r.getTransactionId()));
//                    return new IfmxStreamTransactionRecord(h.beginRecord, h.closingRecord, h.records);
//                }
//                continue;
//            }
//            if (r.getType() != IfmxStreamRecordType.METADATA && r.getType() != IfmxStreamRecordType.TIMEOUT) {
//                this.logger.warn("Missing transaction start for record: {}", r);
//            }
//        }
//
//        return null;
//    }
//
//    private void processRecord(IfmxStreamRecord r, TransactionHolder h) {
//        if (this.logger.isDebugEnabled()) {
//            this.logger.debug("Processing [{}] record for transaction id: {}", r.getType(), Integer.valueOf(r.getTransactionId()));
//        }
//        if (r.hasOperationData() && this.dataFilterSet.contains(r.getType())) {
//            h.records.add((IfmxStreamOperationRecord) r);
//        } else if (r.getType() == IfmxStreamRecordType.COMMIT || r.getType() == IfmxStreamRecordType.ROLLBACK) {
//
//            if (this.transactionFilterSet.contains(r.getType()) && (
//                    !h.records.isEmpty() || (h.records.isEmpty() && this.sendEmptyTransactions))) {
//                h.closingRecord = r;
//
//            } else {
//
//                this.transactionMap.remove(Integer.valueOf(r.getTransactionId()));
//            }
//
//        } else if (r.getType() == IfmxStreamRecordType.DISCARD) {
//            this.logger.debug("Processing DISCARD record");
//            Iterator<IfmxStreamOperationRecord> i = h.records.iterator();
//            while (i.hasNext()) {
//                IfmxStreamOperationRecord record = i.next();
//                if (record.getSequenceId() >= r.getSequenceId()) {
//                    this.logger.debug("Discarding record with sequence [{}]", Long.valueOf(record.getSequenceId()));
//                    i.remove();
//                }
//            }
//        }
//    }
//
//
//    public IfxTransactionEngine setOperationFilters(IfmxStreamRecordType... types) {
//        for (IfmxStreamRecordType t : types) {
//            if (!ALLOWED_DATA_FILTERS.contains(t)) {
//                throw new IllegalArgumentException("Type [" + t + "] cannot be filtered. Correct options are: " + ALLOWED_DATA_FILTERS);
//            }
//        }
//        this.dataFilterSet = EnumSet.copyOf(Arrays.asList(types));
//        return this;
//    }
//
//
//    public IfxTransactionEngine setTransactionFilters(IfmxStreamRecordType... types) {
//        for (IfmxStreamRecordType t : types) {
//            if (!ALLOWED_TRANSACTION_FILTERS.contains(t)) {
//                throw new IllegalArgumentException("Type [" + t + "] cannot be filtered. Correct options are: " + ALLOWED_TRANSACTION_FILTERS);
//            }
//        }
//        this.transactionFilterSet = EnumSet.copyOf(Arrays.asList(types));
//        return this;
//    }
//
//
//    public IfxTransactionEngine returnEmptyTransactions(boolean returnEmpty) {
//        this.sendEmptyTransactions = returnEmpty;
//        return this;
//    }
//
//
//    public void close() throws IfxStreamException {
//        this.engine.close();
//    }
//
//
//    public void init() throws SQLException, IfxStreamException {
//        this.engine.init();
//    }
//
//
//    public int mapSize() {
//        return this.transactionMap.size();
//    }
//
//
//    public IfmxStreamRecord getRecord() throws SQLException, IfxStreamException {
//        return getTransaction();
//    }
//}
//
//
